package com.learn.sunday.es.compont;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

// 定义订单数据类
@Slf4j
@Data
public class Order {
    public String orderId;
    public int userId;
    public Integer amount;

    public Order() {}

    public Order(String orderId, int userId, Integer amount) {
        this.orderId = orderId;
        this.userId = userId;
        this.amount = amount;
    }
    final static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    public static void main(String[] args) throws Exception {
        // 生成订单数据流
        List<Order> orders = new ArrayList<>();
        orders.add(new Order("1", 1, 100));
        orders.add(new Order("2", 2, 201));
        orders.add(new Order("3", 1, 302));
        orders.add(new Order("4", 2, 406));
        DataStream<Order> orderStream = env.fromCollection(orders);

        //使用KeyBy算子将数据流按照用户ID分区，并对每个用户的订单金额求和
        // 根据用户ID作为Key进行分区
        // 对每个分区的订单金额求和
        orderStream.keyBy(order -> order.userId).sum("amount").print();

        env.execute("test");
        Thread.sleep(1000000000L);

    }
}
